package com.wp.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/*
独立消费者模式
 */

public class KafkaConsumerWithOffsetControl3 {
    public static void main(String[] args) {
        Properties props = new Properties();
        //bootstrap-server
        props.put("bootstrap.servers", "192.168.100.141:9092");
        // 指定消费者组
//        prop.put("group.id", "wp");
        // 指定消费位置: earliest/latest/none
//        props.put("auto.offset.reset", "earliest");
        // 指定消费的key的反序列化方式
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 指定消费的value的反序列化方式
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");
        props.put("session.timeout.ms", "60000");
        props.put("max.poll.interval.ms", "60000");
        final Consumer<String, String> consumer = new KafkaConsumer(props);
        List<PartitionInfo> partitionInfoList = consumer.partitionsFor("chantree-all");
        List<TopicPartition> topicPartitionList = new ArrayList();
        if(partitionInfoList != null){
            for(PartitionInfo partitionInfo : partitionInfoList){
                topicPartitionList.add(new TopicPartition(partitionInfo.topic(),partitionInfo.partition()));
                consumer.assign(topicPartitionList);
            }
        }
        System.out.println(consumer.assignment());
        //设置各个partition 的offset
        for (TopicPartition partition : consumer.assignment()) {
            consumer.seek(partition, 0L);
        }

        final Map<TopicPartition,OffsetAndMetadata> currentOffset = new HashMap();
        for (int i = 0; i <10 ; i++)  {
            System.out.println("=====================================================");
            ConsumerRecords<String, String> records = consumer.poll(1000);
            System.out.println(records.count());
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                currentOffset.put(new TopicPartition(record.topic(),record.partition()),new OffsetAndMetadata(record.offset(),"metadata"));
//                try {
//                    System.out.println("模拟消息处理失败的情况");
//                } catch (Exception e) {
//                    consumer.commitAsync(currentOffset,null);
//                }
//                consumer.commitAsync(currentOffset,new OffsetCommitCallback(){
//
//                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
//                        System.out.println("消费完成："+offsets);
//                    }
//                });
            }
        }
        consumer.close();
    }
}
